package com.dongdongshop.mq;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.dubbo.config.annotation.Service;
import com.dongdongshop.config.DongDongSearchConfig;
import com.dongdongshop.pojo.item.TbItem;
import com.dongdongshop.service.ItemServiceI;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;

@Service
//topic=itempage的mq
@RocketMQMessageListener(topic = "dongdongshop-itempage",consumerGroup = "dongdongshop_es",messageModel = MessageModel.BROADCASTING)//配置消费者信息
public class ESItemMQ implements RocketMQListener<Long[]> {

    @Reference
    private ItemServiceI itemServiceI;

    @Resource
    private RestHighLevelClient restHighLevelClient;
    @Override
    public void onMessage(Long[] longs) {
        try {
            for (Long id : longs) {
                //根据goodsId查询
                List<TbItem> items = itemServiceI.ListItemsByGoodsId(id);
                for (TbItem item : items) {
                    IndexRequest request = new IndexRequest("dongdongshop", "shop");
                    try {
                        String userJson = new ObjectMapper().writeValueAsString(item);
                        request.source(userJson, XContentType.JSON);
                        //执行操作
                        IndexResponse response = restHighLevelClient.index(request, DongDongSearchConfig.COMMON_OPTIONS);
                        //提取有用的响应信息
                        System.out.println(">>>index: " + response);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
            System.out.println("消息消费成功");
        }catch (Exception e){
            e.printStackTrace();
            System.out.println("消费失败");
        }
    }
}
